package com.mjk.common.core.rmq.mq;


import com.mjk.common.core.rmq.EventErrorCode;
import com.mjk.common.core.rmq.annotation.Subscribe;
import com.mjk.common.util.AopTargetUtils;
import com.mjk.common.util.MDCUtils;
import com.mjk.common.util.MapUtils;
import com.mjk.common.util.ObjectUtils;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.aop.support.AopUtils;

import java.lang.management.ManagementFactory;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.*;


/**
 * 
 * @author mjk
 * @date 2019年3月8日
 */
public class SubscribeProxys implements MessageHander {
	private final RabbitAdmin admin;
	private final Map<Integer, List<String>> queueCategorys = new ConcurrentHashMap<>();
	private final Map<Integer, Subscribe> queueCategorySubscribes = new ConcurrentHashMap<>();
	private ThreadPoolExecutor threadPoolExecutor;
	//private BlockingQueue<Runnable> subscribeMsgQeueu;
	private final RmqChannel channel;
	private final Map<Integer, CategoryListenerContainer> containers;
	private final Map<String, EventHandler> handlers;
	private final SimpleMessageConverter messageConverter = new SimpleMessageConverter();
	private  String maxLength = "x-max-length";
	private  String messageTtl="x-message-ttl";
	public SubscribeProxys(RmqChannel channel) {
		this.channel = channel;
		this.admin = new RabbitAdmin(channel.getConnectionFactory());
		this.containers = new HashMap<>();
		this.handlers = new ConcurrentHashMap<>();
		int threadNum = channel.getProperty().getThreads();
		if (threadNum == 0) {
			threadNum = Runtime.getRuntime().availableProcessors() * 3;
		}
		channel.getConnectionFactory().addConnectionListener(new ConnectionListener() {
			@Override
			public void onCreate(Connection connection) {
				// Do nothing
			}

			@Override
			public void onClose(Connection connection) {
				// Do nothing
			}
		});

	}

	public synchronized void start() {
        List<Integer> allPrioritys = new ArrayList<>(this.queueCategorys.keySet());
		Collections.sort(allPrioritys);

		int threadNum = channel.getProperty().getThreads();
		if (threadNum == 0) {
			threadNum = Runtime.getRuntime().availableProcessors() * 3;
		}
		BlockingQueue<Runnable> subscribeMsgQeueu = new PriorityBlockingQueue<>(allPrioritys, threadNum);
		this.threadPoolExecutor = new ThreadPoolExecutor(threadNum, threadNum, 5, TimeUnit.MINUTES, subscribeMsgQeueu,
                (r, executor) -> {
                    try {
                        subscribeMsgQeueu.put(r);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });
		for (int p : allPrioritys) {
			List<String> queueName = this.queueCategorys.get(p);
			int prefetchCount = this.queueCategorySubscribes.get(p).prefetchCount();
			if (prefetchCount <= 0) {
				prefetchCount = threadNum * 3;
			}
			String[] queueNameArray = new String[queueName.size()];
			queueName.toArray(queueNameArray);

            final int size = prefetchCount;
			this.containers.computeIfAbsent(p, k -> new CategoryListenerContainer(p, this.channel, queueNameArray,
					this, size));
		}
		for (CategoryListenerContainer container : this.containers.values()) {
			container.start();
		}
	}

	public int doProxy(Object target) {
		Object proxy = AopTargetUtils.getTarget(target);
		Class<?> clazz = null;
		if (proxy != null) {
			clazz = proxy.getClass();
		}
		if (proxy != null && AopUtils.isCglibProxy(proxy)) {
			clazz = proxy.getClass().getSuperclass();
		}
		while (clazz != null) {
			for (Method method : clazz.getMethods()) {
				Subscribe annotation = method.getAnnotation(Subscribe.class);
				if (annotation != null) {
					registerHandler(target, method, annotation);
				}
			}
			clazz = clazz.getSuperclass();
		}
		return EventErrorCode.SUCCEED;
	}

	private void registerHandler(Object target, Method method, Subscribe annotation) {
		if (annotation != null) {
			Class<?>[] parameterTypes = method.getParameterTypes();
			if (parameterTypes.length != 1) {
				throw new IllegalArgumentException(
						"Method " + method + " has @Subscribe annotation, but requires " + parameterTypes.length
								+ " arguments.  Event handler methods " + "must require a single argument.");
			}
			String eventName = annotation.queueName();
			if (ObjectUtils.isEmpty(eventName)) {
				Class<?> eventType = parameterTypes[0];
				if(eventType.getSuperclass()!=null){
					messageConverter.addAllowedListPatterns(eventType.getSuperclass().getName());
				}
				eventName = eventType.getName();
				messageConverter.addAllowedListPatterns(eventName);
			}


			EventHandler handler = new EventHandler(target, method);
			
			Subscribe.Type[] types = annotation.value();
			for (Subscribe.Type type : types) {
				switch (type) {
				case BROADCAST:
					String broadCastName = registerBroadcast(eventName, annotation);
					this.handlers.put(broadCastName, handler);
					break;
				case QUEUE:
					String queueName = registerQueue(eventName, annotation);
					this.handlers.put(queueName, handler);
					break;
				case P2P:
					String p2pName = registerP2P(eventName, annotation);
					this.handlers.put(p2pName, handler);
					break;
				}
			}
		}
	}

	private String registerP2P(String eventName, Subscribe sbscribe) {
		String queueName = eventName + "_" + ManagementFactory.getRuntimeMXBean().getName();
		Map<String, Object> args = new HashMap<>();
		if (sbscribe.queueMaxLength() > 0) {
			args.put(maxLength, sbscribe.queueMaxLength());
		}
		if (sbscribe.messageTtl() > 0) {
			args.put(messageTtl, sbscribe.messageTtl());
		}
		Queue queue = new Queue(queueName, false, true, true, args);
		this.admin.declareQueue(queue);
        TopicExchange exchange = new TopicExchange(queueName, false, true);
		this.admin.declareExchange(exchange);
		this.admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(queueName));
		addQueue(queueName, sbscribe);
		return queueName;
	}

	private String registerBroadcast(String eventName, Subscribe sbscribe) {
		String queueName = eventName + "_" + ManagementFactory.getRuntimeMXBean().getName()+"broadcast";
		Map<String, Object> args = new HashMap<>();
		if (sbscribe.queueMaxLength() > 0) {
			args.put(maxLength, sbscribe.queueMaxLength());
		}
		if (sbscribe.messageTtl() > 0) {
			args.put(messageTtl, sbscribe.messageTtl());
		}
		Queue queue = new Queue(queueName, false, true, true, args);
		this.admin.declareQueue(queue);
		FanoutExchange exchange = new FanoutExchange(eventName + "_broadcast", false, true);
		this.admin.declareExchange(exchange);
		this.admin.declareBinding(BindingBuilder.bind(queue).to(exchange));
		addQueue(queueName, sbscribe);
		return queueName;
	}

	private String registerQueue(String eventName, Subscribe sbscribe) {
		Map<String, Object> args = new HashMap<>();
		if (sbscribe.queueMaxLength() > 0) {
			args.put(maxLength, sbscribe.queueMaxLength());
		}
		if (sbscribe.messageTtl() > 0) {
			args.put(messageTtl, sbscribe.messageTtl());
		}
		Queue queue = new Queue(eventName, true, false, false, args);
		this.admin.declareQueue(queue);
		TopicExchange exchange = new TopicExchange(eventName, false, true);
		this.admin.declareExchange(exchange);
		this.admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(eventName));
		addQueue(eventName, sbscribe);
		return eventName;
	}

	public void addQueue(String queueName, Subscribe subscribe) {
        List<String> queueNameList = this.queueCategorys.computeIfAbsent(subscribe.priority(), k -> new ArrayList<>());
        if (!queueNameList.contains(queueName)) {
			queueNameList.add(queueName);
		}
		this.queueCategorySubscribes.put(subscribe.priority(), subscribe);
	}

	public int cleanQueue(Class<?> clazz, boolean noWait) {
		return this.cleanQueue(clazz.getName(), noWait);
	}

	public int cleanQueue(String queueName, boolean noWait) {
		this.admin.purgeQueue(queueName, noWait);
		return EventErrorCode.SUCCEED;
	}

	@Override
	public void handleMessage(final Message event, int priority) {
		threadPoolExecutor.execute(new RunnablePriority(priority) {
			@Override
			public void run() {
				String eventName = event.getMessageProperties().getConsumerQueue();
				Object obj = messageConverter.fromMessage(event);				
				EventHandler handler = handlers.get(eventName);
				if (handler != null) {

					Map<String, Object> head= event.getMessageProperties().getHeaders();
					String traceId = MapUtils.getString(head, EventErrorCode.HEADER_TRACE);
					if(!ObjectUtils.isEmpty(traceId)) {
						MDCUtils.setTraceId(traceId);
					}
					handler.handle(eventName, false, null, obj);

				}
			}
		});
	}
}
